package com.lost.octopus.es.processor.mq.consumer.capture;

import com.lost.octopus.es.processor.entity.db.EsDataCaptureRuleEntity;
import com.lost.octopus.es.processor.mq.consumer.dto.SinkDeal;
import com.lost.octopus.es.processor.service.common.EsDataCaptureRuleService;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 数据捕获，配置管理器
 * @author zhangbh
 * @date 2025/5/16 11:45
 */
@Log4j2
@Component
public class DataCaptureConfigManager {

    @Resource
    private EsDataCaptureRuleService esDataCaptureRuleService;

    public void loadConfig(Map<String, Set<SinkDeal>> captureStage) {
        // 从数据库加载配置
        List<EsDataCaptureRuleEntity> list = esDataCaptureRuleService.list();
        Map<String, SinkDeal> dealMap = new HashMap<>();
        List<String> gzList = new ArrayList<>();
        // 先处理parent，否则配置顺序加载比较严格 因为配置很少，所以不会有影响
        for (EsDataCaptureRuleEntity rule : list) {
            if (!rule.isTop()) {
                continue;
            }
            captureStage.putIfAbsent(rule.getKey(), new LinkedHashSet<>());
            Set<SinkDeal> deals = captureStage.get(rule.getKey());
            SinkDeal deal = new SinkDeal(rule, true);
            deals.add(deal);
            dealMap.putIfAbsent(rule.getRuleId(), deal);
        }
        // 处理子规则，如果有特殊要求则设置sort
        for (EsDataCaptureRuleEntity rule : list) {
            gzList.add(rule.getRuleId());
            String sql = rule.getQuerySql();
            boolean noSql = StringUtils.isBlank(sql);
            if (!rule.isTop()) {
                String parentId = rule.getParentId();
                SinkDeal parentDeal = dealMap.get(parentId);
                if (parentDeal == null) {
                    log.error("parentId ={} 注册顺序有误！childrenId={}", rule.getParentId(), rule.getRuleId());
                    throw new IllegalArgumentException(String.format("parentId =%s 注册顺序有误！childrenId=%s", rule.getParentId(), rule.getRuleId()));
                }
                SinkDeal deal = parentDeal.childSink(rule, !noSql);
                dealMap.putIfAbsent(rule.getRuleId(), deal);
            }
        }
        log.info("数据捕获处理到es共存在 {} 个表处理配置，监测表清单={} 规则清单 {}", list.size(), captureStage.keySet(), gzList);
    }
}
